package com.paradogs.framework.core.queue;

import com.paradogs.framework.core.exceptions.GlobalExceptionDispatch;
import com.paradogs.framework.core.msg.PRMsgContextHolder;
import com.paradogs.framework.core.utils.SpringUtils;
import com.paradogs.framework.core.utils.StringUtils;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.InvocationTargetException;
import java.time.LocalDateTime;
import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * @author: yumi
 * @date: 2023/7/27  13:38
 * @Description: TODO
 */
@Slf4j
public abstract class QueueConsumerThread<T> extends Thread {

    /**
     * 处理的队列，一个消费者对应一个队列
     */
    @Setter
    @Getter
    private BlockingQueue<T> queue;

    /**
     * 启动时间
     */
    @Getter
    private LocalDateTime startTime;

    /**
     * 处理的消息总数
     */
    @Getter
    private Long handleCount;

    /**
     * 唯一标识
     */
    @Setter
    private String key;

    /**
     * 是否执行 Tick 循环，相同类型线程只要一个循环即可
     */
    @Getter
    @Setter
    private boolean isLoop = false;

    /**
     * 线程是否正在运行，用于单例防错
     */
    private boolean isRunning = false;

    /**
     * 全局异常处理转发器
     */
    private GlobalExceptionDispatch globalExceptionDispatch = SpringUtils.getBean(GlobalExceptionDispatch.class);


    public QueueConsumerThread() {
        this.startTime = LocalDateTime.now();
        this.handleCount = 0L;
        this.queue = new LinkedBlockingQueue<>();
    }

    /**
     * 处理消息
     */
    public abstract void handle(T msg) throws InvocationTargetException, IllegalAccessException;

    public void handleBefore(T msg) {}
    public void handleAfter(T msg) {}

    /**
     * 循环检测
     */
    public abstract void loop();

    @Override
    public final void run() {

        log.info("[{}] queueConsumerThread start success", getKey());

        while (true) {

            /**
             * 循环检测事件
             */
            if (isLoop) {
                try {
                    loop();
                } catch (Exception e) {
                    log.error("loop error: {}", e.getMessage());
                }
            }

            /**
             * 处理消息，消息队列为空时，等待一段时间再处理，如果有消息则处理完直接找下一个
             */
            T msg = queue.poll();
            if (msg == null) {
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    continue;
                }

            }

            this.handleCount++;
            try {
                handleBefore(msg);
                PRMsgContextHolder.getSnapshot();

                handle(msg);

                handleAfter(msg);

            } catch (Exception e) {
                // 出现异常时，转发到全局异常处理
                globalExceptionDispatch.forward(e);
            }

        }

    }

    public String getKey() {
        if (StringUtils.isBlank(this.key)) {
            this.key = this.getClass().getSimpleName() + "-" + UUID.randomUUID().toString().replace("-", "");
        }
        return this.key;
    }

    public Boolean produce(T data) {
        return this.queue.add(data);
    }

}
